﻿using System;
using System.Collections.Generic;
using System.ComponentModel.Composition;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;

namespace FinMkts.Messaging.Pgm
{
    public class PgmSubscriber : Subscriber
    {
        private readonly object _gate = new object();
        readonly AddressedMessageSerializer _serializer = new AddressedMessageSerializer();
        private Socket _socket;
        private bool _stopped;

        [Import]
        public PgmConfiguration Config { get; set; }

        public override void Start()
        {
            if (Config == null) throw new InvalidOperationException("Config must be set before Start is called");

            // create socket and allow others to subscribe on the same multicast group
            _socket = new Socket(AddressFamily.InterNetwork, SocketType.Rdm, RmConstants.ProtocolType);
            var multicastEndPoint = new IPEndPoint(Config.MulticastAddress, Config.MulticastPort);
            _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
            _socket.Bind(multicastEndPoint);

            // set options before listen
            _socket.SetSocketOption(RmSocketOptionName.HighSpeedIntranet, Config.HighSpeedIntranet);
            _socket.Listen(5);

            // accept incomming connections
            var acceptArgs = new SocketAsyncEventArgs();
            acceptArgs.Completed += AcceptCompleted;
            bool pending = _socket.AcceptAsync(acceptArgs);
            if (!pending)
            {
                AcceptCompleted(_socket, acceptArgs);
            }
        }

        public override void Stop()
        {
            lock (_gate)
            {
                _stopped = true;
                Subject.OnCompleted();
            }
        }

        private void AcceptCompleted(object sender, SocketAsyncEventArgs acceptArgs)
        {
            bool acceptPending;
            do
            {
                if (_stopped) return;
                if (acceptArgs.SocketError != SocketError.Success)
                {
                    Subject.OnError(new SocketException((int) acceptArgs.SocketError));
                    return;
                }

                var receiveSocket = acceptArgs.AcceptSocket;
                Debug.Assert(receiveSocket != null, "receiveSocket is null");
                acceptArgs.AcceptSocket = null;
                receiveSocket.SetSocketOption(RmSocketOptionName.HighSpeedIntranet, Config.HighSpeedIntranet);

                // start receiving on new socket: set up a SocketAsyncEventArgs and new buffer
                var receiveArgs = new SocketAsyncEventArgs();
                var readBuffer = new byte[Config.MaxMessageSize.Bytes];
                receiveArgs.SetBuffer(readBuffer, 0, readBuffer.Length);
                receiveArgs.Completed += ReceiveCompleted;
                var receivePending = receiveSocket.ReceiveAsync(receiveArgs);
                if (!receivePending)
                {
                    ReceiveCompleted(receiveSocket, receiveArgs);
                }

                if (_stopped) return;
                acceptPending = _socket.AcceptAsync(acceptArgs);
            } while (!acceptPending);
        }

        private void ReceiveCompleted(object sender, SocketAsyncEventArgs args)
        {
            bool receivePending;
            do
            {
                switch (args.SocketError)
                {
                    case SocketError.Success:
                        break;
                    case SocketError.ConnectionReset: 
                        // sender as stopped, stop receiving
                        args.Completed -= ReceiveCompleted;
                        return;
                    default:
                        // some other error, stop receiving
                        Subject.OnError(new SocketException((int)args.SocketError));
                        return;
                }

                // the message may contain many messages, deserialize them all
                int offset = 0;
                do
                {
                    var message = _serializer.Deserialize(args.Buffer, ref offset, args.BytesTransferred);
                    if (message == null) break; // no more messages in buffer
                    if (_stopped) return;
                    Subject.OnNext(message);
                } while ((offset + AddressedMessageSerializer.MinMessageSize) < args.BytesTransferred);
                
                // clear the buffer ready for the next receive
                Array.Clear(args.Buffer, 0, args.Buffer.Length);

                var receiveSocket = (Socket) sender;
                if (_stopped) return;
                receivePending = receiveSocket.ReceiveAsync(args);
            } while (!receivePending);
        }

    }
}